<?php
/**
推送模板消息

 **/

$db = new swoole_mysql;
$url = "http://www.4399.com"; //测试用
$cfg = [
    "total_process"=>120, //最高限制进程数
    "limit"=>"0",//数据量限制，为零就是不限制
    "min_id"=>"8549548",//从最小的id开始，用于二次推送
    "worknum"=>100,//默认单个进程处理的请求数量,程序会优先total_process的限制,达到最大进程数时会超过这个数量
    "app_pg_id"=>0//小程序id
];

$msg_quene = [];//消息队列
$server = array(
   'host' => '192.168.3.112',//外网地址：
    'port' => 3306,
   'user' => 'root',
   'password' => 'root',
    'database' => 'console-api',
    'charset' => 'utf8', //指定字符集
    'timeout' => 2,  // 可选：连接超时时间（非查询超时时间），默认为SW_MYSQL_CONNECT_TIMEOUT（1.0）
);
//404    otlG05BepPpr6PjVhDjKQcW6wBWs



$db->connect($server, function ($db, $r) {
    global $cfg;
	$dbPreix = "mj_";

    if ($r === false) {
        var_dump($db->connect_errno, $db->connect_error);
        die;
    }
    //获取队列数据
    $sql = 'SELECT wxopen_template_send_queue_id,app_id,app_secret,touser,template_id,page,form_id,data,emphasis_keyword FROM '.$dbPreix.'wxopen_template_send_queue '.
    		'WHERE 1=1 ';
//    		 ' AND status = "1"';
    if($cfg['app_pg_id']){
        $sql .=' AND applications_mini_program_id ="'.$cfg['app_pg_id'].'"';
    }
    if($cfg['min_id']){
        $sql .=' AND wxopen_template_send_queue_id >"'.$cfg['min_id'].'"';
    }
    if ($cfg['limit']){
        $sql .= ' LIMIT '.$cfg['limit'];
    }
    $db->query($sql, function(swoole_mysql $db, $r) {
        global $cfg;
        $dataLen = count($r);
        echo ">>获取数据完毕长度:".$dataLen;
        $worknum = $cfg['worknum']>0?$cfg['worknum']:100;//一个进程处理多少个请求
        if ($r === false) {
            var_dump($db->error, $db->errno);
        } elseif ($r === true ) {

        	//把查询结果放到数组
//        	post_data();
             var_dump($db->affected_rows, $db->insert_id);
        }

        $worknum = $dataLen/$worknum > $cfg['total_process']?ceil($dataLen/$cfg['total_process']):$worknum;
        //多个一组多进程请求
        $quene = [];
        $temp = [];
        for($i = 1;$i<=$dataLen ;$i++){
            //设置序号
            $r[$i-1]['no'] = $i;
            $temp[] = $r[$i-1];
            if($i%$worknum == 0){
                $quene[] = $temp;
                $temp = [];
            }
        }
        //记录最后一条记录
        addLog('LAST_ID:'.$r[$dataLen-1]["wxopen_template_send_queue_id"],"task");
        if($temp){
            $quene[] = $temp;
        }
        multi_process($quene);

//        print_r($quene);
        // var_dump($r);
        $db->close();
    });
});
function multi_process($quene){
    global $msg_quene;
    echo ">>开始分布".count($quene)."个进程";
    //首先赋值到公共内存中
    foreach($quene as $k => $v){
        $msg_quene[$k] = $v;
    }

    foreach($quene as $k => $v){
        $process = new swoole_process('doProcess');//创建新进程
        $pid = $process->start();//启动进程，返回pid
        $process->write($k);//向管道内写入当前i的值
        $workers[$pid] = $process;//放入进程池中
    }
//    print_r($msg_quene);
}



/**
 * 多进程回调数据处理
 * @param swoole_process $worker
 */
function doProcess(swoole_process $worker){
    global $url;//获取全局的URL，因为是共享内存的
    global $msg_quene;
//    echo ">>进程启动了";
    $i = $worker->read();//读取i值

    $data = $msg_quene[$i];
    if($data){
        foreach ($data as $k=>$val){
//            $result = file_get_contents($url);
//            print_r($val);
            $result = send_weapp_msg($val['app_id'], $val['app_secret'], $val['touser'], $val['template_id'], $val['page'], $val['form_id'], json_decode($val['data']), $val['emphasis_keyword']);
//            print_r($result);
            if($result['errcode'] != 0){
                addLog("<ID:".$val['wxopen_template_send_queue_id']."> ".$val['touser'].json_encode($result,true),"fail");
            }else{
                addLog("<ID:".$val['wxopen_template_send_queue_id']."> ".$val['touser'].json_encode($result,true),"success");
            }

            echo "pid:".$worker->pid." ".date('md H:i:s')." 异步数据".$val['no']."：".json_encode($result,true)."\n";
//            echo "pid:".$worker->pid." ".date('md H:i:s')." 异步数据".$val['no']."\n";
        }
    }
//    $worker->write("pid :". $worker->pid.' res: '.$res.PHP_EOL);//往管道里面写数据 pipe 将curl的结果再传回主进程
    // $res = curlTest($url[$i], (10-$i*2));//执行curl操作

    // echo "写入信息: pid: ".$worker->pid.' res: '.$res.' './*$worker->callback.*/PHP_EOL;
    //sleep(5);
    //$worker->exit(0); 退出子进程
}

/**
 * 获取服务器上的token文件
 * @param $appid
 * @return null
 */
function get_accessToken($appid, $appsecret)
{
    $domain = "https://cc.men.com";
    $url = $domain."/Message/Task/getAccessToken/appid/".$appid."/appsecret/".$appsecret;
//    echo $url;
    $res = file_get_contents($url);
    if($res){
        return $res;
    }else{
        return null;
    }
}


/**
 * https的异步请求
 * @param $url 域名
 * @param $request url参数
 * @param $data 请求变量
 * @param string $callback 回调参数
 * @return bool
 */
function http_post2($url, $request, $data, $callback = '')
{
    Swoole\Async::dnsLookup($url, function ($domainName, $ip) use($request,$data, $callback) {
        echo $ip;
        $cli = new swoole_http_client($ip, 80);
        $cli->setHeaders([
            'Host' => $domainName,
            "User-Agent" => 'Chrome/49.0.2587.3',
            'Accept' => 'text/html,application/xhtml+xml,application/xml',
            'Accept-Encoding' => 'gzip',
        ]);
        $cli->post($request,$data, function ($cli) use ($callback) {
            if($callback){
                $callback($cli->body);
            }
            echo "Length: " . strlen($cli->body) . "\n";
            echo $cli->body;
        });
    });
    return true;
}
/**
 * @param $appid
 * @param $appsecret
 * @param $touser 接收者（用户）的 openid
 * @param $template_id  所需下发的模板消息的id
 * @param $page 点击模板卡片后的跳转页面，仅限本小程序内的页面。支持带参数,（示例index?foo=bar）。该字段不填则模板无跳转。
 * @param $form_id 表单提交场景下，为 submit 事件带上的 formId；支付场景下，为本次支付的 prepay_id
 * @param $data  模板内容，不填则下发空模板
 * @param $emphasis_keyword   模板需要放大的关键词，不填则默认无放大
 * @return Ambigous <boolean, multitype:>
 * 小程序发送模板消息
 */

function send_weapp_msg($appid='',$appsecret='',$touser='',$template_id='',$page='',$form_id='',$data=array(),$emphasis_keyword=NULL) {
    $access_token = get_accessToken($appid, $appsecret);
//    echo "发送微信数据";
    $url = 'https://api.weixin.qq.com/cgi-bin/message/wxopen/template/send?access_token='.$access_token;
    $send=array();
//    $touser = 'otlG05BepPpr6PjVhDjKQcW6wBWs';
//    $form_id = '1524132992663';
    $send['touser'] = $touser;
    $send['template_id'] = $template_id;
    $send['page']=$page;
    $send['form_id'] = $form_id;
    $send['data']=$data;
    $send['emphasis_keyword']=$emphasis_keyword;
    $result = http_post($url,json_encode($send));
    //好像更慢了，决定弃用下面的异步http client
//    $result = http_post2('api.weixin.qq.com','/cgi-bin/message/wxopen/template/send?access_token='.$access_token,$send,function($result) use($send){
//        $result = json_decode($result,true);
//        if($result['errcode'] === 0){
//            addLog("<FORM_ID:". $send['form_id']."> ".$send['touser'].json_encode($result,true),"success");
//        }else{
//            addLog("<FROM_ID:". $send['form_id']."> ".$send['touser'].json_encode($result,true),"fail");
//        }
//    });
    if($result){
        $res =json_decode($result,true);
        return $res;
    }else{
        return false;
    }

}
/**
 * 日志记录
 * @param $data 记录
 */
function addLog($data, $fileName)
{
    $log_file = "$fileName.log";
//    $content =var_export($data,TRUE);
    $header = '>>>>>>' . date('Y-m-d H:i:s') . '<<<<<'."\r\n";
    $content = $header.$data;
    $content .= "\r\n\n";
    file_put_contents($log_file,$content, FILE_APPEND);
    return true;
}
/**

 * POST 请求

 * @param string $url

 * @param array $param

 * @param boolean $post_file 是否文件上传

 * @return string content

 */

function http_post($url,$param,$post_file=false){

    $oCurl = curl_init();

    curl_setopt($oCurl, CURLOPT_SAFE_UPLOAD, false);

    if(stripos($url,"https://")!==FALSE){

        curl_setopt($oCurl, CURLOPT_SSL_VERIFYPEER, FALSE);

        curl_setopt($oCurl, CURLOPT_SSL_VERIFYHOST, false);

        curl_setopt($oCurl, CURLOPT_SSLVERSION, 1); //CURL_SSLVERSION_TLSv1

    }

    if (is_string($param) || $post_file) {

        $strPOST = $param;

    } else {

        $aPOST = array();

        foreach($param as $key=>$val){

            $aPOST[] = $key."=".urlencode($val);

        }

        $strPOST =  join("&", $aPOST);

    }
    curl_setopt($oCurl, CURLOPT_URL, $url);

    curl_setopt($oCurl, CURLOPT_RETURNTRANSFER, 1 );

    curl_setopt($oCurl, CURLOPT_POST,true);

    curl_setopt($oCurl, CURLOPT_POSTFIELDS,$strPOST);

    $sContent = curl_exec($oCurl);

    $aStatus = curl_getinfo($oCurl);

    curl_close($oCurl);

    if(intval($aStatus["http_code"])==200){

        return $sContent;

    }else{

        return false;

    }

}